package bblgw

import (
	"bblcloud/bblvolume"
	"encoding/json"
	"fmt"
	"github.com/facebookgo/httpcontrol"
	"io"
	"io/ioutil"
	"net/http"
	"net/http/httputil"
	"net/url"
	"strings"
	"sync"
	"time"
)

const (
	KiB           = 1024
	MiB           = 1024 * KiB
	GiB           = 1024 * MiB
	PUT_FILE_SIZE = 4 * MiB
)

type StoreClient struct {
	http_client *http.Client
	ms_server   string
	ms_master   string
	ms_slave    []string
	open_objs   map[string]string
	_lock       *sync.Mutex
}

func NewStoreClient(master string, slave []string) *StoreClient {
	return &StoreClient{
		http_client: &http.Client{
			Transport: &httpcontrol.Transport{
				DialTimeout:           30 * time.Second,
				RequestTimeout:        0 * time.Second,
				ResponseHeaderTimeout: 0 * time.Second,
			},
		},
		ms_master: master,
		ms_slave:  slave,
		ms_server: master,
		open_objs: make(map[string]string),
		_lock:     new(sync.Mutex),
	}
}

const (
	OPEN          = "OPEN"
	APPEND        = "APPEND"
	CLOSE         = "CLOSE"
	READ          = "READ"
	WRITE         = "WRITE"
	DELETE        = "DELETE"
	TRUNCATE      = "TRUNCATE"
	STAT          = "STAT"
	RENAME        = "RENAME"
	LINK          = "LINK"
	UNLINK        = "UNLINK"
	ACK_WRITE     = "ACK_WRITE"
	ACK_APPEND    = "ACK_APPEND"
	ADD_VOLUME    = "ADD_VOLUME"
	REMOVE_VOLUME = "REMOVE_VOLUME"
	ADD_CLIENT    = "ADD_CLIENT"
	REMOVE_CLIENT = "REMOVE_CLIENT"
	JOIN          = "JOIN"
	LEAVE         = "LEAVE"
	GET_JOB       = "GET_JOB"
	ACK_JOB       = "ACK_JOB"
	CLUSTER_STAT  = "CLUSTER_STAT"
)

func (c *StoreClient) url(path string) string {
	if strings.HasPrefix(path, "/") {
		return fmt.Sprintf("http://%v%v", c.ms_server, path)
	}
	return fmt.Sprintf("http://%v/%v", c.ms_server, path)
}

type Stat struct {
	Size  int64 `json:"size"`
	CTime int64 `json:"ctime"`
	MTime int64 `json:"mtime"`
	UID   int   `json:"uid"`
	GID   int   `json:"gid"`
	Perm  int   `json:"perm"`
	Crc32 int64 `json:"crc32"`
}

func (c *StoreClient) Stat(path string) (stat *Stat, err error) {
	data, err, _ := c.get_content(STAT, path, nil)
	if err != nil {
		return nil, err
	}
	stat = new(Stat)
	err = json.Unmarshal(data, stat)
	return stat, err
}

func (c *StoreClient) WriteProxy(rw http.ResponseWriter, req *http.Request) {

}

func (c *StoreClient) ReadProxy(rw http.ResponseWriter, r *http.Request) {
	path := r.URL.Path
	data, err, resp := c.get_content(READ, path, nil)
	if err != nil {
		log.Error(err)
		badRequest(rw, r)
		return
	}
	log.Debug(string(data))
	var vols []string
	err = json.Unmarshal(data, &vols)
	if err != nil {
		log.Error(err)
		badRequest(rw, r)
		return
	}

	mod := int(Hash(fmt.Sprintf("%s-%d", path, Rand())) % uint32(len(vols)))
	if resp.Header.Get("X-LINK-TO") != "" {
		path = resp.Header.Get("X-LINK-TO")
	}
	targetUrl, err := url.Parse(fmt.Sprintf("http://%v", vols[mod]))
	if err != nil {
		log.Error(err)
		badRequest(rw, r)
	}
	r.RequestURI = path
	r.URL.Path = path
	r.Method = READ
	log.Debug("ReadProxy Request", r)
	proxy := httputil.NewSingleHostReverseProxy(targetUrl)
	proxy.ServeHTTP(rw, r)
	log.Debug("ReadProxy Response", rw)
}
func (c *StoreClient) ReadToWriter(path string, w io.WriteCloser) (written int64, err error) {
	data, err, resp := c.get_content(READ, path, nil)
	if err != nil {
		return 0, err
	}
	log.Debug(string(data))
	var vols []string
	err = json.Unmarshal(data, &vols)
	if err != nil {
		return 0, err
	}
	mod := int(Hash(fmt.Sprintf("%s-%d", path, Rand())) % uint32(len(vols)))
	if resp.Header.Get("X-LINK-TO") != "" {
		path = resp.Header.Get("X-LINK-TO")
	}
	reqn, err := c.NewRequest(READ, fmt.Sprintf("http://%v%v", vols[mod], path), nil)
	if err != nil {
		return 0, err
	}
	respn, err := c.http_client.Do(reqn)
	if err != nil {
		return 0, err
	}
	defer respn.Body.Close()
	log.Debug("READ\t", reqn, "\t\n", respn)
	if respn.StatusCode == 200 || respn.StatusCode == 206 {
		return io.Copy(w, respn.Body)
	}
	return 0, fmt.Errorf("%v", resp.Status)

}
func (c *StoreClient) ACK_Written(path string, written int64) error {
	req, err := c.NewRequest(ACK_WRITE, path, nil)
	if err != nil {
		return err
	}
	req.Header.Add("X-WRITTEN-SIZE", fmt.Sprintf("%v", written))
	resp, err := c.http_client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	log.Debug("ACK\t", req, "\t\n", resp)
	if resp.StatusCode == 200 {
		return nil
	}
	return fmt.Errorf("%v", resp.Status)
}

type multirequest struct {
	w           *io.PipeWriter
	r           *io.PipeReader
	url         string
	wait        *sync.WaitGroup
	http_client *http.Client
}

func new_multireq(url string, http_client *http.Client, wait *sync.WaitGroup) *multirequest {
	r, w := io.Pipe()
	return &multirequest{r: r, w: w, url: url, wait: wait, http_client: http_client}
}
func (r *multirequest) Write(b []byte) (int, error) {
	return r.w.Write(b)
}
func (r *multirequest) Close() error {
	return r.w.Close()
}

func (r *multirequest) do() error {
	defer r.wait.Done()
	req, err := http.NewRequest(WRITE, r.url, r.r)
	if err != nil {
		return err
	}
	req.Header.Set("User-Agent", "bblcloud-client/1.1")
	req.Header.Set("Content-Type", "application/octet-stream")
	//req.ContentLength = size
	resp, err := r.http_client.Do(req)
	if err != nil {
		return err
	}

	defer resp.Body.Close()
	log.Debug("WRITE\t", req, "\t\n", resp)
	if resp.StatusCode != 200 {
		return fmt.Errorf(resp.Status)
	}
	return nil
}

func (c *StoreClient) WriteFromReader(f io.ReadCloser, size int64, path string) (written int64, err error) {
	data, err, _ := c.get_content(WRITE, path, nil)
	if err != nil {
		return 0, err
	}
	log.Debug(string(data))
	var vols []string
	err = json.Unmarshal(data, &vols)
	if err != nil || len(vols) == 0 {
		return 0, err
	}
	log.Debug(vols)
	writers := []io.Writer{}
	wait := new(sync.WaitGroup)

	for _, vol := range vols {
		mreq := new_multireq(fmt.Sprintf("http://%v%v", vol, path), c.http_client, wait)
		writers = append(writers, mreq)
		wait.Add(1)
		go mreq.do()
	}
	written, err = io.Copy(io.MultiWriter(writers...), f)
	for _, w := range writers {
		w.(*multirequest).Close()
	}
	if err != nil {
		log.Error("io.Copy(io.MultiWriter(writers...), f)", err)
		return written, err
	}
	wait.Wait()
	err = c.ACK_Written(path, written)
	log.Debug("WriteFromReader", written, err)
	return written, err
}

func (c *StoreClient) Delete(path string) (err error) {
	req, err := c.NewRequest(DELETE, path, nil)
	if err != nil {
		return err
	}
	resp, err := c.http_client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	log.Debug("DELETE\t", req, "\t\n", resp)
	if resp.StatusCode == 200 {
		return nil
	}
	return fmt.Errorf("%v", resp.Status)
}

func (c *StoreClient) Truncate(path string) (err error) {
	req, err := c.NewRequest(TRUNCATE, path, nil)
	if err != nil {
		return err
	}
	resp, err := c.http_client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	log.Debug("TRUNCATE\t", req, "\t\n", resp)
	if resp.StatusCode == 200 {
		return nil
	}
	return fmt.Errorf("%v %v", resp.StatusCode, resp.Status)
}

func (c *StoreClient) Rename(from, to string) (err error) {
	req, err := c.NewRequest(RENAME, from, nil)
	if err != nil {
		return err
	}
	req.Header.Add("X-RENAME-TO", to)
	resp, err := c.http_client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	log.Debug("RENAME\t", req, "\t\n", resp)
	if resp.StatusCode == 200 {
		return nil
	}
	return fmt.Errorf("%v %v", resp.StatusCode, resp.Status)
}

func (c *StoreClient) Link(from, to string) (err error) {
	req, err := c.NewRequest(LINK, from, nil)
	if err != nil {
		return err
	}
	req.Header.Add("X-LINK-TO", to)
	resp, err := c.http_client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	log.Debug("LINK\t", req, "\t\n", resp)
	if resp.StatusCode == 200 {
		return nil
	}
	return fmt.Errorf("%v %v", resp.StatusCode, resp.Status)
}

func (c *StoreClient) Unlink(path string) (err error) {
	req, err := c.NewRequest(UNLINK, path, nil)
	if err != nil {
		return err
	}
	resp, err := c.http_client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	log.Debug("UNLINK\t", req, "\t\n", resp)
	if resp.StatusCode == 200 {
		return nil
	}
	return fmt.Errorf("%v %v", resp.StatusCode, resp.Status)
}

func (c *StoreClient) AddVolume(uuid string) (err error) {
	req, err := c.NewRequest(ADD_VOLUME, uuid, nil)
	if err != nil {
		return err
	}
	resp, err := c.http_client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	//log.Debug("ADD_VOLUME\t", req, "\t\n", resp)
	if resp.StatusCode == 200 {
		return nil
	}
	return fmt.Errorf("%v %v", resp.StatusCode, resp.Status)
}

func (c *StoreClient) RemoveVolume(uuid string) (err error) {
	req, err := c.NewRequest(REMOVE_VOLUME, uuid, nil)
	if err != nil {
		return err
	}
	resp, err := c.http_client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	//log.Debug("REMOVE_VOLUME\t", req, "\t\n", resp)
	if resp.StatusCode == 200 {
		return nil
	}
	return fmt.Errorf("%v %v", resp.StatusCode, resp.Status)
}

func (c *StoreClient) AddClient(addr, perm string) (err error) {
	req, err := c.NewRequest(ADD_CLIENT, addr+","+perm, nil)
	if err != nil {
		return err
	}
	resp, err := c.http_client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	//log.Debug("ADD_VOLUME\t", req, "\t\n", resp)
	if resp.StatusCode == 200 {
		return nil
	}
	return fmt.Errorf("%v %v", resp.StatusCode, resp.Status)
}

func (c *StoreClient) RemoveClient(addr string) (err error) {
	req, err := c.NewRequest(REMOVE_CLIENT, addr, nil)
	if err != nil {
		return err
	}
	resp, err := c.http_client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	//log.Debug("REMOVE_VOLUME\t", req, "\t\n", resp)
	if resp.StatusCode == 200 {
		return nil
	}
	return fmt.Errorf("%v %v", resp.StatusCode, resp.Status)
}

type ClusterStat struct {
	TotalSpace int64 `json:",omitempty"`
	UserSpace  int64 `json:",omitempty"`
	Status     int
	Version    int
	Volumes    []struct {
		UUID               string    //"550e8400-e29b-41d4-a716-446655440000",
		Name               string    //Storage#1
		Address            string    //"192.168.1.1:2003",
		TotalSpace         uint64    `json:",omitempty"`
		UsedSpace          uint64    `json:",omitempty"`
		Export             string    //"/export/storage",
		MountDev           string    `json:",omitempty"`
		Status             int       //"0:Offline,1:Online,2:Maintenance,3:Rebalance"
		HeartbeatTime      time.Time `json:",omitempty"`
		MalfunctionNum     int
		LastMalfunctionNum time.Time          `json:",omitempty"`
		Stat               *bblvolume.VolStat `json:",omitempty"`
	}
	Clients []struct {
		Address string
		Perm    string
	}
}

func (c *StoreClient) NewRequest(method, path string, body io.ReadCloser) (req *http.Request, err error) {
	if strings.HasPrefix(path, "http://") {
		req, err = http.NewRequest(method, path, body)
	} else {
		req, err = http.NewRequest(method, c.url(path), body)
	}
	req.Header.Set("User-Agent", "bblcloud-client/1.1")
	return req, err
}

func (c *StoreClient) get_content(method, path string, body io.ReadCloser) ([]byte, error, *http.Response) {
	req, err := c.NewRequest(method, path, body)
	if err != nil {
		return nil, err, nil
	}
	resp, err := c.http_client.Do(req)
	if err != nil {
		return nil, err, nil
	}

	defer resp.Body.Close()
	if resp.StatusCode != 200 {
		return nil, fmt.Errorf("%v", resp.Status), resp
	}
	b, err := ioutil.ReadAll(resp.Body)
	return b, err, resp
}
func (c *StoreClient) ClusterStat() (stat *ClusterStat, err error) {
	req, err := c.NewRequest(CLUSTER_STAT, "/", nil)
	if err != nil {
		return nil, err
	}
	resp, err := c.http_client.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	//log.Debug("CLUSTER_STAT\t", req, "\t\n", resp)
	if resp.StatusCode == 200 {
		b, err := ioutil.ReadAll(resp.Body)
		if err != nil {
			return nil, err
		}
		cstat := ClusterStat{}
		err = json.Unmarshal(b, &cstat)
		return &cstat, err
	}
	return nil, fmt.Errorf("%v %v", resp.StatusCode, resp.Status)
}

func (c *StoreClient) Open(path string) (fd string, err error) {
	req, err := c.NewRequest(WRITE, path, nil)
	if err != nil {
		return "", err
	}
	resp, err := c.http_client.Do(req)
	if err != nil {
		return "", err
	}
	log.Debug("WRITE\t", req, "\t\n", resp)
	if resp.StatusCode != 200 {
		return "", fmt.Errorf("%v %v", resp.StatusCode, resp.Status)
	}
	data, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return "", err
	}
	var vols []string
	err = json.Unmarshal(data, &vols)
	if err != nil {
		return "", err
	}
	if len(vols) == 0 {
		return "", fmt.Errorf("%v %v", 404, "Not Found")
	}
	vol_url := fmt.Sprintf("http://%v%v", vols[0], path)
	req, err = http.NewRequest(OPEN, vol_url, nil)
	if err != nil {
		return "", err
	}
	resp, err = c.http_client.Do(req)
	if err != nil {
		return "", err
	}
	log.Debug("OPEN\t", req, "\t\n", resp)
	if resp.StatusCode == 200 {
		fd = UUID()
		c._lock.Lock()
		c.open_objs[fd] = vol_url
		c._lock.Unlock()
		return fd, nil
	}
	return "", fmt.Errorf("%v %v", resp.StatusCode, resp.Status)
}
func (c *StoreClient) ACK_Append(path string, written int64) error {
	req, err := c.NewRequest(ACK_APPEND, path, nil)
	if err != nil {
		return err
	}
	req.Header.Add("X-WRITTEN-SIZE", fmt.Sprintf("%v", written))
	resp, err := c.http_client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	log.Debug("ACK\t", req, "\t\n", resp)
	if resp.StatusCode == 200 {
		return nil
	}
	return fmt.Errorf("%v %v", resp.StatusCode, resp.Status)
}
func (c *StoreClient) Append(fd string, body io.ReadCloser, size int64) error {
	c._lock.Lock()
	vol_url, exist := c.open_objs[fd]
	if !exist {
		c._lock.Unlock()
		return fmt.Errorf("%v %v", 404, "Not Open")
	}
	c._lock.Unlock()
	req, err := http.NewRequest(APPEND, vol_url, body)
	if err != nil {
		return err
	}
	req.Header.Add("Content-Type", "application/octet-stream")
	req.ContentLength = size
	resp, err := c.http_client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	log.Debug("APPEND\t", req, "\t\n", resp)
	if resp.StatusCode == 200 {
		return nil
	}
	return fmt.Errorf("%v %v", resp.StatusCode, resp.Status)
}

func (c *StoreClient) Close(fd string) error {
	c._lock.Lock()
	vol_url, exist := c.open_objs[fd]
	if !exist {
		c._lock.Unlock()
		return fmt.Errorf("%v %v", 404, "Not Open")
	}
	delete(c.open_objs, fd)
	c._lock.Unlock()
	req, err := http.NewRequest(CLOSE, vol_url, nil)
	if err != nil {
		return err
	}
	resp, err := c.http_client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	log.Debug("CLOSE\t", req, "\t\n", resp)
	if resp.StatusCode == 200 {
		return nil
	}
	return fmt.Errorf("%v %v", resp.StatusCode, resp.Status)
}
